package com.huawei.spare.part.replace.rx;

import io.reactivex.Observable;
import java.util.concurrent.atomic.AtomicInteger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class Retry {

  private static final Logger log = LoggerFactory.getLogger(Retry.class);

  public static void main(String[] args) throws InterruptedException {
    retryWhen();
  }

  private static void retry() throws InterruptedException {
    Observable.create(emitter -> {
      log.info("on run");
      emitter.onError(new RuntimeException("error"));
    }).retry(3).subscribe();

    Thread.sleep(10000L);
  }

  private static void retryWhen() throws InterruptedException {
    final RetryWithDelay handler =
        new RetryWithDelay(5, 1000,
            "[%d] Failed, reason: %s");
    final RetryWithDelay handler2 =
        new RetryWithDelay(5, 1000,
            "[%d] Failed, reason: %s");

    AtomicInteger times = new AtomicInteger(0);
    AtomicInteger times2 = new AtomicInteger(0);
    Observable.create(emitter -> {
      log.info("on run");
      if (times.incrementAndGet() <= 4) {
        emitter.onError(new RuntimeException(times + " error"));
      } else {
        emitter.onNext(true);
      }
    }).retryWhen(handler).flatMap(observable -> Observable.defer(() -> {
          if (times2.incrementAndGet() <= 4) {
            return Observable.error(new RuntimeException(times2 + " failed"));
          } else {
            return Observable.just(true);
          }
        }).retryWhen(handler2)
    ).doOnSubscribe((d) -> {
      log.info("[{}] Start task");
    }).doOnDispose(() -> {
      log.info("[%s] Cancel task now");
    }).doOnError((e) -> {
      log.info("[%s] Execute task failed");
    }).doOnComplete(() -> {
      log.info("[{}] Execute task succeed");
    }).subscribe((n) -> {
      log.info("completed: " + n);
    }, e -> {
      log.info("failed " + e);
    }, () -> {
      log.info("[{}] Execute task succeed");
    });

    Thread.sleep(20000L);
  }
}
